package com.demo.concurrent;

import java.util.ArrayList;
import java.util.HashSet;
import java.util.List;
import java.util.Set;
import java.util.concurrent.*;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.AtomicLong;
import java.util.concurrent.atomic.AtomicReference;
import java.util.concurrent.locks.Lock;
import java.util.concurrent.locks.ReadWriteLock;
import java.util.concurrent.locks.ReentrantLock;
import java.util.concurrent.locks.ReentrantReadWriteLock;

/**
 * @author lee9213@163.com
 * @version 1.0
 * @date 2017-05-07 13:24
 */
public class QueueDemo {

    public static void main(String[] args) {
        LinkedBlockingQueue queue = new LinkedBlockingQueue();
        try {
            Object o = queue.take();
            System.out.println(o);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }
    public static void maina(String[] args) throws Exception {
        //链式阻塞队列
        LinkedBlockingQueue queue = new LinkedBlockingQueue();
        queue.add(1);//抛异常
        queue.offer(true);//特定值，返回true或false
        queue.put(1); //阻塞
        queue.offer(1, 1, TimeUnit.SECONDS);//等待超时，返回特定值

        //数组阻塞队列，内部采用数组
        ArrayBlockingQueue array = new ArrayBlockingQueue(10);
        array.put("1");

        //延时队列,内部采用PriorityQueue
        DelayQueue delay = new DelayQueue();
        delay.put(new DelayedElement());

        //优先级阻塞队列，内部采用数组
        PriorityBlockingQueue priority = new PriorityBlockingQueue();
        priority.put(1);

        //同步队列，只能有一个元素
        SynchronousQueue synchronous = new SynchronousQueue();
        synchronous.put(1);

        //双端链式阻塞队列
        LinkedBlockingDeque deque = new LinkedBlockingDeque();
        deque.putFirst(1);
        deque.putLast(1);
        deque.addFirst(1);

        deque.addLast(1);
        deque.offerLast(1);
        deque.offerFirst(1);

        //并发Map
        ConcurrentMap map = new ConcurrentHashMap();
        map.remove("", "");

        //并发导航映射
        ConcurrentNavigableMap navigableMap = new ConcurrentSkipListMap();
        map.put("1", "one");
        map.put("2", "two");
        map.put("3", "three");
        navigableMap.headMap("2"); //小与2的map
        navigableMap.tailMap("2");//大于等于2的map
        navigableMap.subMap("2", "3");//大于等于2，小于3

        //闭锁
        CountDownLatch countDownLatch = new CountDownLatch(3);
        countDownLatch.await();
        countDownLatch.countDown();

        //栅栏
        CyclicBarrier cyclicBarrier = new CyclicBarrier(3);
        cyclicBarrier.await();
        cyclicBarrier.await(1, TimeUnit.SECONDS);

        //交换机
        Exchanger exchanger = new Exchanger();
        exchanger.exchange("1");

        //信号量
        Semaphore semaphore = new Semaphore(2);
        semaphore.acquire();
        semaphore.release();

        //创建线程池
        //定长线程池，可控制最大并发数，超出线程队列中等待
        ExecutorService service = Executors.newFixedThreadPool(5);
        //可缓存的线程池，线程池无限大
        Executors.newCachedThreadPool();
        //延时或定时线程池
        ScheduledExecutorService scheduledService = Executors.newScheduledThreadPool(5);
        //延时5秒执行
        ScheduledFuture scheduledFuture = scheduledService.schedule((Callable<Object>) () -> "1", 5, TimeUnit.SECONDS);
        System.out.println(scheduledFuture.get());
        //每隔多久执行一次，如果前一个任务未完成等待前一个任务完成之后再执行
//        scheduledService.scheduleAtFixedRate();
        //两个任务间隔多少时间执行
//        scheduledService.scheduleWithFixedDelay();
        //同时只有一个线程执行
        Executors.newSingleThreadExecutor();
        //同时只有一个线程执行，定时任务
        Executors.newSingleThreadScheduledExecutor();
        //jdk1.8提供，返回ForkJoinPool
        Executors.newWorkStealingPool();

        service.execute(() -> {
        });

        //线程执行完成，无返回值，正确完成返回null
        Future future = service.submit(() -> {
        });
        future.get();

        //线程执行完成，有返回值
        Future future1 = service.submit((Callable<Object>) () -> "1");
        future1.get();

        Set<Callable<Object>> callables = new HashSet<>();
        //返回其中任意一个
        Object object = service.invokeAny(callables);
        //返回全部结果
        List<Future<Object>> futureList = service.invokeAll(callables);
        //遍历结果
        futureList.forEach(objectFuture -> {
            try {
                objectFuture.get();
            } catch (InterruptedException e) {
                e.printStackTrace();
            } catch (ExecutionException e) {
                e.printStackTrace();
            }
        });

        //ForkJoinPool
        ForkJoinPool forkJoinPool = new ForkJoinPool();
        MyRecursiveAction myRecursiveAction = new MyRecursiveAction(128);
        forkJoinPool.invoke(myRecursiveAction);

        MyRecursiveTask myRecursiveTask = new MyRecursiveTask(128);
        Long result = forkJoinPool.invoke(myRecursiveTask);
        System.out.println("result = " + result);

        //同步锁
        Lock lock = new ReentrantLock();
        lock.lock();//获取lock锁，阻塞状态
        lock.lockInterruptibly();//获取lock锁，可中断
        lock.tryLock();//试图立即获取lock锁，如果锁定成功，返回true，否则返回false
        lock.tryLock(1,TimeUnit.SECONDS);//试图立即获取锁，超时状态
        lock.unlock();//解锁

        //读写锁
        ReadWriteLock readWriteLock = new ReentrantReadWriteLock();
        readWriteLock.readLock().lock();
        readWriteLock.readLock().unlock();

        readWriteLock.writeLock().lock();
        readWriteLock.writeLock().unlock();

        AtomicBoolean atomicBoolean = new AtomicBoolean();
        AtomicInteger integer = new AtomicInteger();
        AtomicLong atomicLong = new AtomicLong();

        AtomicReference reference = new AtomicReference();





    }

    //没有返回值的任务
    static class MyRecursiveAction extends RecursiveAction {
        private long workLoad = 0;

        public MyRecursiveAction(long workLoad) {
            this.workLoad = workLoad;
        }

        @Override
        protected void compute() {
            System.out.println("Splitting workLoad : " + this.workLoad);
            //if work is above threshold, break tasks up into smaller tasks
            if (this.workLoad > 16) {
                List<MyRecursiveAction> subtasks = new ArrayList<>();
                subtasks.addAll(createSubtasks());
                subtasks.forEach(subtask -> subtask.fork());

            } else {
                System.out.println("Doing workLoad myself: " + this.workLoad);
            }
        }

        public List<MyRecursiveAction> createSubtasks() {
            List<MyRecursiveAction> subtasks = new ArrayList<>();
            MyRecursiveAction subtask1 = new MyRecursiveAction(this.workLoad / 2);
            MyRecursiveAction subtask2 = new MyRecursiveAction(this.workLoad / 2);
            subtasks.add(subtask1);
            subtasks.add(subtask2);
            return subtasks;
        }
    }

    //有返回值的任务
    static class MyRecursiveTask extends RecursiveTask<Long> {

        private long workLoad = 0;

        public MyRecursiveTask(long workLoad) {
            this.workLoad = workLoad;
        }

        @Override
        protected Long compute() {
            if (this.workLoad > 16) {
                System.out.println("Splitting workLoad : " + this.workLoad);
                List<MyRecursiveTask> subtasks = new ArrayList<>();
                subtasks.addAll(createSubtasks());

                subtasks.forEach(subtask -> subtask.fork());
                long result = 0;
                for (MyRecursiveTask subtask : subtasks) {
                    result += subtask.join();
                }
                return result;
            } else {
                System.out.println("Doing workLoad myself: " + this.workLoad);
                return this.workLoad * 3;
            }
        }

        public List<MyRecursiveTask> createSubtasks() {
            List<MyRecursiveTask> subtasks = new ArrayList<>();
            subtasks.add(new MyRecursiveTask(this.workLoad / 2));
            subtasks.add(new MyRecursiveTask(this.workLoad / 2));
            return subtasks;
        }
    }
}
